package com.ubox.mqtt_upush.mqttbroker;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import javax.annotation.Resource;

import org.apache.activemq.broker.ConnectionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.alibaba.fastjson.JSONObject;
import com.ubox.mqtt_upush.config.MqttBrokerConfig;
import com.ubox.mqtt_upush.dao.ClientOfPhoneRepository;
import com.ubox.mqtt_upush.dao.PhoneRepository;
import com.ubox.mqtt_upush.domain.ClientOfPhone;
import com.ubox.mqtt_upush.domain.MKey;
import com.ubox.mqtt_upush.domain.Phone;
import com.ubox.mqtt_upush.mqttclient.FuseMQTTClientProvider;
import com.ubox.mqtt_upush.utils.JsonUtil;

/**
 * 接收到的消息进行按消息类型处理，查找对应该的回调处理器进行处理
 * 
 * @Author:fanglin
 * @ClassName:OrangeTemplate
 * @Date:2017年3月23日 下午6:39:22
 */

@Service
public class MessageDispatcher {

	private final static Logger log = LoggerFactory.getLogger(MessageDispatcher.class);

	private final Map<String, MessageCallback<Object>> callbackMap = new ConcurrentHashMap<>(1000);
	/** 默认阻塞请求超时时间,18s */
	private final int defaultTimeout = 18 * 1000;
	/** 唯一实例，使用volatile保证了多线程访问时template变量的可见性，避免初始化时其他变量属性还没赋值完时，被另外线程调用 */

	@Resource
	FuseMQTTClientProvider client;
	@Resource
	ClientOfPhoneRepository cp;
	@Resource
	PhoneRepository pdao;

	/** 注册收到指定消息key时，回调绑定对象进行处理 在相应callback对象初始化时进行调用注册 */
	public void bind(String key, MessageCallback<Object> callback) {
		if (!callbackMap.containsKey(key)) {
			callbackMap.put(key, callback);
			log.info("绑定[key:{}]回调接口完成", key);
		} else if (!callbackMap.get(key).getClass().getName().equals(callback.getClass().getName())) {
			log.error("不允许相同{}绑定不同回调接口:{}", key, callback.getClass().getName());
		}
	}

	/** 解除key对应的回调绑定 */
	public void unbind(String key) {
		if (callbackMap.containsKey(key)) {
			Object callback = callbackMap.remove(key);
			log.info("解绑[key:{}]的回调接口:{}", key, callback.getClass().getName());
			callback = null;
		}
	}

	/** 判断指定消息标识是否有回调 */
	public boolean hasCallback(String key) {
		if (callbackMap.containsKey(key)) {
			return true;
		}
		return false;
	}

	/** 获取指定消息标识的回调 */
	public MessageCallback<Object> getCallback(String key) {
		if (callbackMap.containsKey(key)) {
			return callbackMap.get(key);
		}
		return null;
	}

	public void sendMsg(String clientId, String user, String msg, ConnectionContext session) throws Exception {
		if (!client.isConnected()) {
			client.connect();
		}
		client.publish("v1/cloud/" + clientId + "/response", msg.getBytes(), MqttBrokerConfig.AT_LEAST_ONCE, false);
	}

	/**
	 * 分发器，将收到的消息进行回调或分发，将结果(非空)响应到发送方 此方法由mina的接收到消息的方法调用处理消息
	 * message为消息对象，一般为json字符串 按张实定义的机器升级协议中，消息中必须包含"type"、"data"、"cn"、"vmID"
	 * 
	 * @Author:fanglin
	 * @Date:2017年3月28日 下午2:09:46
	 * @param session
	 * @param message
	 * @throws Exception
	 */
	@Async
	public void dispatcher(ConnectionContext session, String clientId, Object message) throws Exception {
		/** 1、将消息转为JSONObject对象方便解析，消息对象为String类型的JSON串 */
		JSONObject jsonObject = null;
		try {
			if (message instanceof String) {
				jsonObject = JSONObject.parseObject((String) message);
			} else {
				log.error("[vmID:{},sId:{}]收到的消息非String类型:{}", session.getClientId(),
						session.getConnectionId().getValue(), message.getClass().getName());
				return;
			}
		} catch (Exception e) {
			jsonObject = new JSONObject();
			jsonObject.put(MKey.R_CODE, MKey.PARAM_ERR);
			jsonObject.put(MKey.R_MSG, "消息内容非json格式！");
			jsonObject.put(MKey.FIELD_TYPE, MKey.TYPE_R);
			log.error("消息内容非json格式！{}", message);
			this.sendMsg(clientId, clientId, jsonObject.toJSONString(), session);
			return;
		}
		/** 2、获取消息类型 */
		Integer type = jsonObject.getInteger(MKey.FIELD_TYPE);
		if (type == null) {
			jsonObject.put(MKey.R_CODE, MKey.PARAM_ERR);
			jsonObject.put(MKey.R_MSG, "消息中缺少[" + MKey.FIELD_TYPE + "]属性！");
			jsonObject.put(MKey.FIELD_TYPE, MKey.TYPE_R);

			this.sendMsg(clientId, clientId, jsonObject.toJSONString(), session);
			log.error(jsonObject.toString());
			return;
		}
		/** 6、业务消息分发处理打印，记录日志 */
		// log.info("业务消息分发处理：{}", jsonObject.toString());

		if (type.intValue() == MKey.BIND) {
			ClientOfPhone p = JsonUtil.toBean(jsonObject.toJSONString(), ClientOfPhone.class);
			p.setCreateTime(new Date());
			try {
				ClientOfPhone query = cp.findBycentral(p.getCentral());
				if (query != null) {
					cp.delete(query.getId());
				}
				cp.save(p);

			} catch (Exception e) {
				log.error("BIND error", e);
			}

			List<ClientOfPhone> list = cp.findByUser(p.getUser());

			List<String> centrals = new ArrayList<>();
			for (ClientOfPhone clientOfPhone : list) {
				centrals.add(clientOfPhone.getCentral());
			}

			JSONObject res = new JSONObject();
			res.put(MKey.FIELD_TYPE, type.intValue());
			res.put(MKey.R_CODE, 0);
			res.put(MKey.R_MSG, "success");
			res.put(MKey.CENTRALS, centrals);
			this.sendMsg(clientId, p.getUser() + "", res.toJSONString(), session);
		}

		else if (type.intValue() == MKey.UNBIND) {
			ClientOfPhone p = JsonUtil.toBean(jsonObject.toJSONString(), ClientOfPhone.class);
			p.setCreateTime(new Date());
			ClientOfPhone d = cp.findBycentral(p.getCentral());
			if (d != null) {
				cp.delete(d.getId());
			}

			List<ClientOfPhone> list = cp.findByUser(p.getUser());

			List<String> centrals = new ArrayList<>();
			for (ClientOfPhone clientOfPhone : list) {
				centrals.add(clientOfPhone.getCentral());
			}

			JSONObject res = new JSONObject();
			res.put(MKey.FIELD_TYPE, type.intValue());
			res.put(MKey.R_CODE, 0);
			res.put(MKey.R_MSG, "success");
			res.put(MKey.CENTRALS, centrals);
			this.sendMsg(clientId, p.getUser() + "", JsonUtil.toString(res), session);
		}

		else if (type.intValue() == MKey.LOGIN) {
			Phone p = JsonUtil.toBean(jsonObject.toJSONString(), Phone.class);

			List<ClientOfPhone> list = cp.findByUser(p.getUser());
            Phone user=pdao.findByUser(p.getUser());
            
            if(user==null || !user.getPassword().equals(p.getPassword()))
            {
            	JSONObject res = new JSONObject();
    			res.put(MKey.FIELD_TYPE, type.intValue());
    			res.put(MKey.R_CODE, 201);
    			res.put(MKey.R_MSG, "用户名或者密码错误");
    			this.sendMsg(clientId, p.getUser() + "", res.toJSONString(), session);
            	return;
            }
            
			List<String> centrals = new ArrayList<>();
			for (ClientOfPhone clientOfPhone : list) {
				centrals.add(clientOfPhone.getCentral());
			}

			JSONObject res = new JSONObject();
			res.put(MKey.FIELD_TYPE, type.intValue());
			res.put(MKey.R_CODE, 0);
			res.put(MKey.R_MSG, "success");
			res.put(MKey.CENTRALS, centrals);
			this.sendMsg(clientId, p.getUser() + "", JsonUtil.toString(res), session);
		} else if (type.intValue() == MKey.SMS_CODE) {
			Phone p = JsonUtil.toBean(jsonObject.toJSONString(), Phone.class);
			//SendSmsResponse response=AliMsg.sendSms(p.getUser(),p.getType());
			 
			JSONObject res = new JSONObject();
			res.put(MKey.FIELD_TYPE, type.intValue());
			res.put(MKey.R_CODE, 0);
			res.put(MKey.R_MSG, "success");
			this.sendMsg(clientId, p.getUser() + "", res.toJSONString(), session);
		} else if (type.intValue() == MKey.REGISTERE || type.intValue() == MKey.RESET_PASSWORD) {
			Phone p = JsonUtil.toBean(jsonObject.toJSONString(), Phone.class);
			 
			Phone old = pdao.findByUser(p.getUser());
			if (old != null) {
				p.setId(old.getId());
			}
			pdao.save(p);
			JSONObject res = new JSONObject();
			res.put(MKey.FIELD_TYPE, type.intValue());
			res.put(MKey.R_CODE, 0);
			res.put(MKey.R_MSG, "success");
			this.sendMsg(clientId, p.getUser() + "", res.toJSONString(), session);
		}

	}

	/**
	 * 同步发送消息，不重试
	 * 
	 * @Author:fanglin
	 * @Date:2017年4月25日 上午11:22:39
	 * @param session
	 *            连接会话
	 * @param jsonObject
	 *            消息对象
	 * @param timeout
	 *            超时时间，单位秒，默认18秒
	 * @return
	 * @throws Exception
	 */
	public JSONObject doHandler(ConnectionContext session, JSONObject message, int timeout) throws Exception {
		return doHandler(session, message, timeout, false);
	}

	/**
	 * 同步发送消息
	 * 
	 * @Author:fanglin
	 * @Date:2017年4月25日 上午11:22:39
	 * @param session
	 *            连接会话
	 * @param jsonObject
	 *            消息对象
	 * @param timeout
	 *            超时时间，单位秒，默认18秒
	 * @param isRetry
	 *            是否重试，如果重试，只重试一次
	 * @return
	 * @throws Exception
	 */
	public JSONObject doHandler(ConnectionContext session, JSONObject message, int timeout, boolean isRetry)
			throws Exception {
		// 定义返回结果
		final Object[] result = { null };
		/** 1、session是否有效 **/
		if (session == null) {
			message.put(MKey.R_CODE, MKey.SESSION_ERR);
			message.put(MKey.R_MSG, "机器网络连接异常");
			log.error(message.toString());
			return message;
		}
		/** 2、检查消息中是否包含必要的字段 **/
		Object type = message.get(MKey.FIELD_TYPE);
		if (null == type) {
			message.put(MKey.R_CODE, MKey.ERR);
			message.put(MKey.R_MSG, "消息中缺少[" + MKey.FIELD_TYPE + "]字段！");
			log.error(message.toString());
			return message;
		}
		Object cn = message.get(MKey.FIELD_CN);
		if (null == cn) {
			message.put(MKey.R_CODE, MKey.ERR);
			message.put(MKey.R_MSG, "消息中缺少[" + MKey.FIELD_CN + "]字段！");
			log.error(message.toString());
			return message;
		}

		String innerCode = session.getClientId();
		/** 2.1如果消息类型是以_r结尾，使用异步发送消息 ***/
		if (type instanceof String && ((String) type).endsWith("_r")) {

			this.sendMsg(session.getUserName(), innerCode, message.toJSONString(), session);

			message.put(MKey.R_CODE, MKey.OK);
			message.put(MKey.R_MSG, "消息类型'_r'结尾，使用异步发送消息，发送成功！");
			log.info("[vmID:{},sId:{}]异步发送消息：{}", session.getClientId(), session.getConnection().getConnectionId(),
					message.toString());
			return message;
		}
		/** 3、绑定匿名回调，等待客户端处理完成将结果返回，返回的消息类型加_r后缀，在消息分发处理里调用此匿名回调停止阻塞继续处理 **/
		final String key = type + "_r_" + cn;
		try {
			bind(key, new MessageCallback<Object>() {
				@Override
				public Object callback(JSONObject message) {
					result[0] = message;
					log.info("同步阻塞匿名回调处理 {}", message.toJSONString());
					synchronized (key) {
						key.notifyAll();
					}
					return "停止阻塞[" + key + "]消息完成";
				}
			});
			/** 4、阻塞时间为0将使用默认时间18s */
			timeout = timeout == 0 ? defaultTimeout : timeout;

			/** 5、发送消息开始等待处理 */
			log.info("[vmID:{},sId:{}]同步发送消息：{}", session.getClientId(), session.getConnection().getConnectionId(),
					message.toString());
			this.sendMsg(session.getUserName(), innerCode, message.toJSONString(), session);
			log.info("[vmID:{},sId:{},key:{}]开始等待", session.getClientId(), session.getConnection().getConnectionId(),
					key);
			synchronized (key) {
				key.wait(timeout);
			}
			log.info("[vmID:{},sId:{},key:{}]等待结束", session.getClientId(), session.getConnection().getConnectionId(),
					key);
			// 结果为空，并且有重试的情况
			if (isRetry && result[0] == null) {
				log.info("[vmID:{},sId:{},key:{}]处理结果为空，重试，开始等待", session.getClientId(),
						session.getConnection().getConnectionId(), key);
				this.sendMsg(session.getUserName(), innerCode, message.toJSONString(), session);
				synchronized (key) {
					key.wait(timeout);
				}
				log.info("[vmID:{},sId:{},key:{}]重试等待结束", session.getClientId(),
						session.getConnection().getConnectionId(), key);
			}
			/** 6、处理返回结果 */
			if (result[0] == null) {
				message.put(MKey.R_CODE, MKey.RESP_TIMEOUT);
				message.put(MKey.R_MSG, "消息推送失败，响应超时！");
				log.warn(message.toString());
			} else {
				message = (JSONObject) result[0];
				message.put(MKey.R_CODE, MKey.OK);
				message.put(MKey.R_MSG, "消息推送成功");
				log.info(message.toString());
			}
		} catch (Exception e) {
			message.put(MKey.R_CODE, MKey.ERR);
			message.put(MKey.R_MSG, "同步发送消息异常！" + e.getClass().getName() + ":" + e.getMessage());
			log.error(message.toString());
		} finally {
			/** 处理完成解绑匿名的回调处理器 */
			unbind(key);
		}
		return message;
	}

}
